package com.linesum.code.kafka;

import java.util.Properties;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.junit.Test;

/**
 * 生产者
 * @author chenzf
 * @date 2017年6月9日上午10:13:02
 */
@SuppressWarnings("static-method")
public class KafkaTestProducer {
	private final KafkaProducer<String, String> producer;

	public KafkaTestProducer() {
		Properties props = new Properties();
		props.put("bootstrap.servers", "localhost:9092");// kafka的端口
		props.put("acks", "all"); //ack方式，all，会等所有的commit最慢的方式
		props.put("retries", 0); //失败是否重试，设置会有可能产生重复数据
		props.put("batch.size", 16384); //对于每个partition的batch buffer大小
		props.put("linger.ms", 1); //等多久，如果buffer没满，比如设为1，即消息发送会多1ms的延迟，如果buffer没满
		props.put("buffer.memory", 33554432); //整个producer可以用于buffer的内存大小
		props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
		props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

		producer = new KafkaProducer<String, String>(props);
	}

	public void produce() {
		int messageNo = 0;
		final int COUNT = 1;
		while (messageNo < COUNT) {
			String key = String.valueOf(messageNo);
			String data = "hello kafka message " + key;
			ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>(KafkaTopic.TEST_TOPICE,
					key, data);
			producer.send(producerRecord);
			System.out.println(data);
			messageNo++;
		}
		producer.close();
	}

	@Test
	public void send() {
		new KafkaTestProducer().produce();
	}
}
